package com.hanxiaozhang.zookeeper;

import com.hanxiaozhang.Configuration;
import com.hanxiaozhang.DistributedLock;
import com.hanxiaozhang.DistributedLockFactoryBuilder;
import com.hanxiaozhang.util.DistributedLockUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;


/**
 * 功能描述: <br>
 * 〈zk清楚无效节点〉
 * <p>
 * 产生原因：
 * Zk不允许在临时节点下创建子节点，InterProcessMutex工具类会根据加锁时，
 * 传入path的创建一个持久节点，然后在这个持久节点下创建顺序临时节点。如果，
 * 不定期清理，就会导致Zk节点数量将会急速递增。
 * <p>
 * 清除频率：
 * 默认每小时清除一次，单位时间秒。清除一天前的
 *
 * @Author:hanxinghua
 * @Date: 2022/9/2
 */
@Slf4j
public class ZkClearInvalidNodeThread extends Thread {


    public final static String LOCK_ID = "ZkClearInvalidNodeThread_lock";

    private CuratorFramework client;

    private Configuration config;


    public ZkClearInvalidNodeThread(CuratorFramework client, Configuration config) {
        super.setDaemon(true);
        super.setName("ZkClearInvalidNodeThread");
        this.client = client;
        this.config = config;
    }

    @Override
    public synchronized void run() {
        while (true) {
            ZkDistributedLock lock = null;
            Boolean hasLock = false;
            try {
                // 增加随机数，避免不同启动者冲突
                Random rd = new Random();
                int random = rd.nextInt(30);
                Thread.currentThread().wait((config.getZkClearInvalidNodeRate() + random) * 1000);
                // 加锁  
                lock = new ZkDistributedLock(LOCK_ID, client, config);
                hasLock = lock.tryLock(1L, TimeUnit.SECONDS);
                if (hasLock) {
                    clearInvalidNode();
                } else {
                    log.error("zk分布式锁清除无效节点，获取锁失败，本次清除结束!");
                }
            } catch (Exception e) {
                log.error("zk分布式锁清除无效节点异常: ", e);
            } finally {
                if (hasLock) {
                    try {
                        lock.unlock();
                    } catch (Exception e) {
                        log.error("释放锁异常", e);
                    }
                }
            }
        }

    }

    /**
     * 删除无效节点
     *
     * @throws Exception
     */
    private void clearInvalidNode() throws Exception {

        String zkBaseLockPath = config.getZkBaseLockPath();
        if (zkBaseLockPath.endsWith(DistributedLockUtil.SEPARATOR)) {
            zkBaseLockPath = zkBaseLockPath.substring(zkBaseLockPath.length() - 1);
        }
        if (client.checkExists().forPath(zkBaseLockPath) == null) {
            log.error("没有使用zk分布式锁,清除结束,当前路径是：{}", zkBaseLockPath);
            return;
        }
        List<String> paths = client.getChildren().forPath(zkBaseLockPath);
        Date date = new Date();
        String currentPath = "";
        for (String path : paths) {
            try {
                Stat stat = new Stat();
                client.getData().storingStatIn(stat).forPath(zkBaseLockPath + path);
                if (stat.getMtime() < (date.getTime() - (config.getZkClearInvalidNodeBeforeTime() * 1000))
                        && stat.getNumChildren() == 0) {
                    currentPath = zkBaseLockPath + DistributedLockUtil.SEPARATOR + path;
                    client.delete().forPath(currentPath);
                    log.info("清除无效节点: {}", currentPath);
                }
            } catch (Exception e) {
                log.error("清除无效节点失败: ", e);
            }
        }
    }

}
